package com.k2data.etl.datasender;

import com.k2data.etl.KafkaClient;
import jodd.io.FileUtil;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.Random;
import java.util.concurrent.TimeUnit;

public class SgDataSender {

    private static String TOPIC_SG_TCS_TEST = "all-tsdb-raw";
    private static String TOPIC_SG_YY_TEST = "yysg_k2";
    private static String TOPIC_SG_S8000_TEST = "sg_s8000_raw_test_1";
        private static String KAFKA_HOST = "117.39.60.34:9093";
//    private static String KAFKA_HOST = "192.168.130.85:9092";
//private static String KAFKA_HOST = "106.120.241.179:19092,106.120.241.180:19092";

    public static void main(String[] args) throws Exception {
        KafkaClient kafkaClient = new KafkaClient("", KAFKA_HOST);

        boolean run = true;

//        String[] lines = FileUtil.readLines(new File("/home/vv/share/lzx.txt"));
//        String[] lines = FileUtil.readLines(new File("/home/vv/share/json_data.json"));

//        BufferedReader fr = new BufferedReader(new FileReader(new File("/home/vv/share/json_data.json")));

//        while (run) {
        String line = null;
        int count = 0;
//        while((line = fr.readLine()) != null){

        // send s8k
//            kafkaClient.setTopicName(TOPIC_SG_S8000_TEST);
//            String msg = S8kDataBuilder.build();
//            System.out.println(kafkaClient.getTopicName() + "-" + msg);
//            kafkaClient.sendMessageAsync(new Random(10000000).nextInt() + "", msg);

        kafkaClient.setTopicName(TOPIC_SG_YY_TEST);
        String msg = TcsDataBuilder.build();

        msg = "{\"tenant\":\"default_tenant\",\"user\":\"\",\"tstable\":\"sg_tcs_ttt_1\",\"timestamp\":1537857124932,\"tags\":{\"factory_id\":\"1\",\"equipment_id\":\"10000\",\"machine_id\":\"1\",\"datatype\":\"1\",\"channel_type\":\"10\",\"channel_id\":\"1\"},\"fields\":{\"niandu\":1.1,\"midu\":33.139,\"youpin\":40.973,\"youwen\":15.606}}";

        System.out.println(kafkaClient.getTopicName() + "-" + msg);
        kafkaClient.sendMessageAsync("1", msg);
//            kafkaClient.sendMessageAsync("1", msg);

//            TimeUnit.SECONDS.sleep(1);

//            break;
        count++;
        if (count % 2000 == 0) {
            System.out.println(count);
        }
//        }
//
//        System.out.println(count);
//
//        fr.close();
    }

    public static void main1(String[] args) throws Exception {
        KafkaClient kafkaClient = new KafkaClient("", KAFKA_HOST);

//        String[] lines = FileUtil.readLines(new File("/home/vv/share/json_data.json"));
        String[] lines = FileUtil.readLines(new File("/home/vv/share/lzx.txt"));

        for (String line : lines) {
            System.out.println(line);

            kafkaClient.setTopicName(TOPIC_SG_TCS_TEST);
            kafkaClient.sendMessageAsync(new Random(10000000).nextInt() + "", line);
        }


    }
}
